package my.thesis.bolts.apicallsReliable;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
import java.net.URLConnection;
import java.util.Map;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

public class CountryDetectorReliable extends BaseRichBolt
{
	private static final long serialVersionUID = 1L;
	private static final String MAPQUESTKEY = "Fmjtd%7Cluur256tng%2Cas%3Do5-9w72lz";
	private static final boolean callapi=false;
	private OutputCollector collector;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		this.collector=collector;
	}
	
	public void execute(Tuple tuple) 
	{
		if(callapi)
		{			
			Long id=tuple.getLongByField("id");
			double lat = tuple.getDoubleByField("lat");
			double lon = tuple.getDoubleByField("lon");
			String text=tuple.getStringByField("text");
	
		    if(lat != 0.0 && lon != 0.0)
			{		    	
		    	String result;
	    		result= GetCountryByLatLon(lat,lon);
		    	
				if(result!=null)
				{
					//2-Digit ISO code for the country
					collector.emit(tuple,new Values(id,result,text));
					collector.ack(tuple);
				}
				else
				{
					 System.out.println("Country not found...");
					 collector.fail(tuple);
				}
			}
		}
		else
		{
			Long id=tuple.getLongByField("id");
			String text=tuple.getStringByField("text");
			String country=tuple.getStringByField("country");
			collector.emit(new Values(id,country,text));
			
			collector.emit(tuple,new Values(id,country,text));			
			collector.ack(tuple);
		}
	}	
	
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		declarer.declare(new Fields("id","country","text"));		
	}
		
	//BOOK TWITTER DATA ANALYTICS - ΑΥΤΟ ΤΟ ΕΧΩ ΠΡΟΣΑΡΜΟΣΕΙ ΑΡΚΕΤΑ
    //Translates a location string to coordinates
	public String GetCountryByLatLon(Double lat,Double lon)
	{
		 //test browser -- IT WORKS!
		 //http://open.mapquestapi.com/geocoding/v1/reverse?key=Fmjtd%7Cluur256tng%2Cas%3Do5-9w72lz&callback=renderReverse&location=29.897126,-95.694483
		
		 String url="http://open.mapquestapi.com/geocoding/v1/reverse?key="+MAPQUESTKEY+"&callback=renderReverse&location="+lat+","+lon+"&outFormat=json";
		 String page = ReadHTML(url);
		 
         if(page!=null)
         {
             try
             {
            	page = page.substring(page.lastIndexOf("(") + 1);
            	page = page.substring(0, page.indexOf(')'));

            	JSONObject pageJson = new JSONObject(page);
            	JSONArray results = pageJson.getJSONArray("results");
            	
            	JSONObject locationsObj = results.getJSONObject(0);
            	JSONArray locationsArr = locationsObj.getJSONArray("locations");
            	
            	if(locationsArr.length()>0)
                {             	   
            	   String country = new String(locationsArr.getJSONObject(0).getString("adminArea1"));
            	   return country;
                } 
             }
             catch(JSONException ex)
             {
            	 System.out.println("Something went wrong: " + ex.getMessage());
             }
         }
         
         return null;
	}
		
	//BOOK TWITTER DATA ANALYTICS
	public String ReadHTML(String url)
    {
        URLConnection conn = null;
        URL theURL = null;
        try
        {
                theURL = new URL(url);
        }
        catch ( MalformedURLException e)
        {
                System.out.println("Bad URL: " + theURL);
                return null;
        }
        String page = "";
        try
        {
        	//Returns a URLConnection instance that represents a connection to the remote object referred to by the URL.
        	//It should be noted that a URLConnection instance does not establish the actual network connection on creation. 
        	//This will happen only when calling URLConnection.connect().
        	//If for the URL's protocol (such as HTTP or JAR), there exists a public, specialized URLConnection subclass belonging to one of the following packages or one 
        	//of their subpackages: java.lang, java.io, java.util, java.net, the connection returned will be of that subclass. For example, for HTTP an HttpURLConnection will 
        	//be returned, and for JAR a JarURLConnection will be returned.
        	
            conn = theURL.openConnection();
            HttpURLConnection huc = (HttpURLConnection) conn;
            conn.setConnectTimeout(2000);
          
            conn.connect();
            
            if(huc.getResponseCode()>=400 && huc.getResponseCode()<=404)
            {
                 return null;
            }
            
            BufferedReader bRead = new BufferedReader(new InputStreamReader((InputStream) conn.getContent()));
            String temp=null;
            while( (temp= bRead.readLine())!=null)
              {
                  page = page+"\n"+temp;
              }
              bRead.close();
        }
        catch (IOException e) 
        {
                return null;
        }
        
        return page;
    }

}
